package dse.messages.multiversion;

import java.util.Hashtable;
import java.util.LinkedList;
import java.util.TreeMap;
import java.util.TreeSet;

import dse.Cluster;
import dse.DataObject;
import dse.Experiment;
import dse.ExperimentRecorder;
import dse.Node;
import dse.Version;
import dse.messages.Message;
import dse.messages.MessagePurpose;
import dse.messages.MessageUtil;
import dse.messages.Transaction;
import dse.messages.singleversion.ShareDownloader;

public class GetTwoLevelDhtShareTransaction extends Transaction implements
		ShareDownloader, VersionReader {
	private Node requester;
	private DataObject desiredObject;
	private Cluster currentCluster;
	
	private LinkedList<Cluster> residentClusters;
	private boolean failed;
	private TreeMap<Version, Hashtable<Integer, LinkedList<Node>>> sharesDownloaded;
	private Hashtable<Integer, LinkedList<Cluster>> failedShares;
	private Version version;
	private int validShareCount;
	private int maxFailure;
	private int successCount;
	private int failureCount;
	//private int processedUpdateMessage;
	private int processedSuccessUpdateMessage;
	private int preOffset;
	private int round;
	
	
	public GetTwoLevelDhtShareTransaction (
			Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
				
		this.requester = requester;
		this.desiredObject = desiredObject;
		this.failed = false;
		this.validShareCount = 0;
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.successCount = 0;
		this.failureCount = 0;
		//this.processedUpdateMessage = 0;
		this.processedSuccessUpdateMessage = 0;
		this.preOffset = 0;
		this.failedShares = null;
		this.round = 1;
	}
	
	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		residentClusters = MessageUtil.getClustersThatMayHaveDataObject(experiment.network(), requester, desiredObject, experiment.modules().assigner());	
		currentCluster = residentClusters.removeFirst();
		Hashtable<Node, Integer> nodes = experiment.modules().assigner().hostedShares(currentCluster, desiredObject);
		
		if(isUpdate)
		{
			recorder.dataUpdateInitiated(super.id, requester, desiredObject, startTime);
			
			recorder.debug(this, "start", startTime);
		
			validShareCount = desiredObject.shareCount();
			version = new Version(requester.id(), requester.sequence());
			for (Node n : nodes.keySet()) {
				sendUpdateShareRequest(recorder, n, nodes.get(n).intValue(), version, startTime);
			}
		}
		else
		{
			recorder.dataRequestInitiated(super.id, requester, desiredObject, startTime);
			
			recorder.debug(this, "start", startTime);
		
			int srf = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject)
						+ this.maxFailure*2;
			
			this.validShareCount = srf>desiredObject.shareCount()? desiredObject.shareCount(): srf;
				
			for (Node n : nodes.keySet()) {
				sendGetLatestShareRequest(recorder, n, startTime);
			}
		}
	}
	
	private void sendGetLatestShareRequest(ExperimentRecorder recorder, Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestLatestShareV2Message(requester, node, desiredObject, 
				MessagePurpose.DataObjectRequest, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendGetPreviousShareRequest(ExperimentRecorder recorder, int n, 
			Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestPreviousShareV2Message(requester, node, desiredObject, 
				n, MessagePurpose.DataObjectRequest, this, round);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendUpdateShareRequest(ExperimentRecorder recorder, Node node, int shareIndex, Version version, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createShareUpdateV4Message(requester, node, desiredObject, shareIndex, 
				version, true, MessagePurpose.DataObjectUpdate, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}

	@Override
	public boolean isDoneDownloadingShares() {
		// TODO Auto-generated method stub
		return failed || finished;
	}

	@Override
	public boolean isDoneUpdatingShares() {
		// TODO Auto-generated method stub
		return failed || finished;
	}

	@Override
	public void shareDownloadFailed(ExperimentRecorder recorder,
			DataObject dobj, int shareIndex, double time) {
		// TODO Auto-generated method stub

	}

	@Override
	public void shareDownloaded(ExperimentRecorder recorder, Node from,
			DataObject dobj, int shareIndex, double time) {
		// TODO Auto-generated method stub

	}

	@Override
	public void shareUpdateFailed(ExperimentRecorder recorder, DataObject dobj,
			int shareIndex, double time) {
		// TODO Auto-generated method stub
		if(isUpdate) {
			if(this.failedShares == null)
				this.failedShares = new Hashtable<Integer, LinkedList<Cluster>>();
			
			if(this.failedShares.get(shareIndex) == null)
			{
				LinkedList<Cluster> lc = new LinkedList<Cluster>();
				lc.addAll(residentClusters);
				this.failedShares.put(shareIndex, lc);
			}
			
			if(this.failedShares.get(shareIndex).size() == 0)
				this.transactionFailed(recorder, time);
			
			Node rn = experiment.modules().assigner().assign(desiredObject, shareIndex, this.failedShares.get(shareIndex).removeFirst());
			this.sendUpdateShareRequest(recorder, rn, shareIndex, version, time);
		}
	}

	@Override
	public void shareUpdated(ExperimentRecorder recorder, Node to,
			DataObject dobj, int shareIndex, double time) {
		// TODO Auto-generated method stub
		if(isUpdate) {		
			this.processedSuccessUpdateMessage++;
			if(this.processedSuccessUpdateMessage >= desiredObject.shareCount())
			{
				transactionSucceed(recorder, time);
			}
		}
	}

	@Override
	public void readVersion(ExperimentRecorder recorder,
			TreeSet<Version> versions, double time, Node from) {
		// TODO Auto-generated method stub

	}
	public void readVersion(ExperimentRecorder recorder, int seq, int index, double time){}


	@Override
	public void readVersion(ExperimentRecorder recorder, Node from,
			Version version, double time) {
		// TODO Auto-generated method stub
		if(finished)
			return;
		if(version == null)
		{
			failureCount++;
			if(failureCount >= desiredObject.shareCount())
			{
				tryNextCluster(recorder, time);
				/*for(Version v : sharesDownloaded.keySet())
				{
					System.out.println("Version:" + v.getClientID() + ":" + v.getSequence() + " : " + sharesDownloaded.get(v).size());
				}
				System.out.println("Failed number exceeds the maximum number");
				transactionFailed(recorder, time);*/
			}
			else if(successCount+failureCount>=desiredObject.shareCount())
				processVersion(recorder, time);
			return;
		}
		if(sharesDownloaded == null)
			sharesDownloaded = new TreeMap<Version, Hashtable<Integer, LinkedList<Node>>>();
		
		
		successCount++;
		int shareIndex = from.getDataShare(desiredObject).getIndex();
		
		if(this.sharesDownloaded.get(version) == null)
		{
			this.sharesDownloaded.put(version, new Hashtable<Integer, LinkedList<Node>>());
			this.sharesDownloaded.get(version).put(shareIndex, new LinkedList<Node>());
			this.sharesDownloaded.get(version).get(shareIndex).add(from);
		}
		else if(this.sharesDownloaded.get(version).get(shareIndex) == null)
		{
			this.sharesDownloaded.get(version).put(shareIndex, new LinkedList<Node>());
			this.sharesDownloaded.get(version).get(shareIndex).add(from);
		}
		else
			this.sharesDownloaded.get(version).get(shareIndex).add(from);
		
		if((successCount+failureCount>=desiredObject.shareCount()))
			processVersion(recorder, time);
	}

	@Override
	public void readVersion(ExperimentRecorder recorder, int seq,
			double time) {
		// TODO Auto-generated method stub

	}
	public void readVersion(ExperimentRecorder recorder, double time, LinkedList<Version> versions, Version version, Node from){}
	
	private void processVersion(ExperimentRecorder recorder, double time)
	{
		if((sharesDownloaded==null) || (sharesDownloaded.size()==0))
			return;
		for(Version version : sharesDownloaded.keySet())
		{
			//System.out.println("Version:" + version.getClientID() + ":" + version.getSequence() + " : " + sharesDownloaded.get(version).size());
			if(sharesDownloaded.get(version).size() == validShareCount)
			{
				transactionSucceed(recorder, time);
				return;
			}
			else if(sharesDownloaded.get(version).size() >= desiredObject.requiredShareCount())
			{
				for(Version ver : sharesDownloaded.keySet())
				{
					if(version.compareTo(ver) != 0)
					{
						for(Integer index : sharesDownloaded.get(version).keySet())
						{
							if(sharesDownloaded.get(ver).get(index) != null)
							{
								for(Node node : sharesDownloaded.get(ver).get(index))
								{
									sendUpdateShareRequest(recorder, node, index, version, time);
								}
							}
						}
					}
				}
				transactionSucceed(recorder, time);	
				return;
			}
		}
		
		failureCount = 0;
		successCount = 0;
		preOffset++;
		if(preOffset == this.maxFailure)
		{
			tryNextCluster(recorder, time);
			return;
		}
		
		this.round++;
		Hashtable<Node, Integer> nodes = experiment.modules().assigner().hostedShares(currentCluster, desiredObject);
		for(Node n : nodes.keySet())
			sendGetPreviousShareRequest(recorder, preOffset, n, time);
	}

	private void tryNextCluster(ExperimentRecorder recorder, double time)
	{
		if(residentClusters.size() == 0)
			transactionFailed(recorder, time);
		else
		{
			currentCluster = residentClusters.removeFirst();
			failureCount = 0;
			successCount = 0;
			preOffset = 0;
			Hashtable<Node, Integer> nodes = experiment.modules().assigner().hostedShares(currentCluster, desiredObject);
			for(Node n : nodes.keySet())
				sendGetLatestShareRequest(recorder, n, time);
		}
	}
	private void transactionSucceed(ExperimentRecorder recorder, double time)
	{
		finished = true;
		failed  = false;
		//System.out.println("Data ID : " + desiredObject.id() + "Version : " + maxVersion.getSequence() + " , " + maxVersion.getClientID());
		if(isUpdate)
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(super.id, requester, desiredObject, true, time);
		}
		else
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataRequestFinished(super.id, requester, desiredObject, true, currentCluster.superNode(),
					false, time);
		}
	}
	
	private void transactionFailed (ExperimentRecorder recorder, double time)
	{
		recorder.debug(this, "transFailed", time);
		
		finished = true;
		failed = true;
		if(isUpdate)
			recorder.dataUpdateFinished(super.id, requester, desiredObject, false, time);
		else
			recorder.dataRequestFinished(super.id, requester, desiredObject, false, null,
				false, time);
	}
	public Transaction getTransaction(){
		return this;
	}
}
